package io.reactivex.rxjava3.core;

import io.reactivex.rxjava3.internal.functions.Functions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicInteger;

public class FlowableDemo {

    private static final Logger log = LoggerFactory.getLogger(FlowableDemo.class);

    public static Flowable<Object> baseFlowable() {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        return
            Flowable
                .create(emitter -> {
                    emitter(emitter, atomicInteger);
                    emitter(emitter, atomicInteger);
                    emitter(emitter, atomicInteger);
                    emitter(emitter, atomicInteger);
                    emitter.onComplete();
                }, BackpressureStrategy.BUFFER);
    }

    private static void emitter(FlowableEmitter<Object> emitter, AtomicInteger atomicInteger) {
        int i = atomicInteger.addAndGet(1);
        log.info("输出:{}", i);
        emitter.onNext(i);
    }

    public static void subscribeFour(Flowable<Object> flowable) {
        subscribeFour(flowable, new AtomicInteger(1));
    }

    public static void subscribeFour(Flowable<Object> flowable, AtomicInteger index) {
        baseSubscribe(flowable, index);
        baseSubscribe(flowable, index);

        final int indexNum = index.getAndAdd(1);
        log.info("第{}次订阅", indexNum);
        //noinspection ResultOfMethodCallIgnored
        flowable.subscribe(o -> {
            log.info("第{}次消费并报错：{}", indexNum, o);
            throw new RuntimeException("主动报错");
        }, e -> log.error("第{}次消费出现异常:{}", indexNum, e.getMessage()), () -> log.info("第{}次订阅完成事件", indexNum));
        baseSubscribe(flowable, index);
    }

    @SuppressWarnings("ResultOfMethodCallIgnored")
    private static void baseSubscribe(Flowable<Object> flowable, AtomicInteger index) {
        final int indexNum = index.getAndAdd(1);
        log.info("第{}次订阅", indexNum);
        flowable.subscribe(o -> logSubscribe(indexNum, o), Functions.ON_ERROR_MISSING, () -> log.info("第{}次订阅完成事件", indexNum));
    }

    private static void logSubscribe(int index, Object o) {
        log.info("第{}次消费：{}", index, o);
    }

}
